package study.nosql.redis.controller;

import cn.hutool.core.map.MapUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.BoundListOperations;
import org.springframework.data.redis.core.BoundStreamOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import study.eggs.result.Result;
import study.eggs.result.Results;
import study.nosql.redis.consumer.StreamQueConsumer;

/**
 * 队列，实现方式
 * （1）list
 * （2）stream
 * （3）pub/sub
 */
@Slf4j
@RestController
@RequestMapping("/queue")
public class _4_QueueController {

    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private StreamQueConsumer streamQueConsumer;

    @Autowired
    private BoundListOperations<String, String> listQueue;
    @Autowired
    private BoundStreamOperations<String, String, Object> streamQueue;

    @RequestMapping("/byList")
    public Result list(String msg) {
        Long result = listQueue.leftPush(msg);
        return Results.ok();
    }

    @RequestMapping("/bySet")
    public Result bySet(String msg) {
        Long result = listQueue.leftPush(msg);
        return Results.ok();
    }

    @RequestMapping("/byPub")
    public Result pub(String msg) {
        redisTemplate.convertAndSend("queue_pub_test", msg);
        return Results.ok();
    }

    @RequestMapping("/byStream")
    public Result stream(String key, String value) {
//        MapRecord<String, String, Object> message = StreamRecords.newRecord()
//                .ofMap(MapUtil.of("", ""))
//                .withStreamKey("queue_stream_test")
////                .withStreamId(StreamOffset.create(null, null))
//                .withValues("field1", "value1",
//                        "field2", "value2");
//
        RecordId id = streamQueue.add(MapUtil.of(key, value));
        log.info("seq= {}, value= {}, timestamp= {}", id.getSequence(), id.getValue(), id.getTimestamp());

        return Results.ok();
    }
}
